/* =========================================
 *           消息存储子服务实现 
 * =========================================*/
#pragma once
#include "channel.hpp"
#include "file.pb.h"
#include "message.hxx"
#include "odb_factory.hpp"
#include "message_dao.hpp"
#include "etcd.hpp"
#include "logger.hpp"
#include "base.pb.h"
#include "message_storage.pb.h"
#include "rabbitmq.hpp"
#include "search_repo.hpp"
#include "user.pb.h"
#include "utils.hpp"
#include <bits/stdint-uintn.h>
#include <boost/date_time/posix_time/ptime.hpp>
#include <brpc/server.h>
#include <cstddef>
#include <cstring>
#include <ctime>
#include <functional>
#include <memory>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <vector>

namespace im_server {

class MsgStorageServiceImpl;
class MsgStorageServer;
class MsgStorageServerBuilder;
using MsgStorageServiceImplPtr = std::shared_ptr<MsgStorageServiceImpl>;
using MsgStorageServerPtr = std::shared_ptr<MsgStorageServer>;
using MsgStorageServerBuilderPtr = std::shared_ptr<MsgStorageServerBuilder>;

class MsgStorageeServiceImpl : public im_proto::MsgStorageService
{
    typedef google::protobuf::Map<std::string, im_proto::FileDownloadData> GPFileMap;
    typedef google::protobuf::Map<std::string, im_proto::UserInfo> GPUserMap;
public:
    MsgStorageeServiceImpl(const shared::dal::ESClientPtr& esClient,
        const shared::dal::OdbPtr& mysqlClient,
        const shared::infra::RpcServiceManagerPtr& rpcServiceMgr,
        const std::string& fileServiceName, 
        const std::string& userServiceName) 
        : esMessage_(std::make_shared<shared::dal::MessageSearch>(esClient))
        , messageDao_(std::make_shared<shared::dal::MessageDao>(mysqlClient))
        , rpcServiceMgr_(rpcServiceMgr)
        , fileServiceName_(fileServiceName) 
        , userServiceName_(userServiceName) { esMessage_->createIndex(); }
    ~MsgStorageeServiceImpl() {}

    // 获取最近N条历史消息
    void GetRecentHisMsg(::google::protobuf::RpcController* controller,
                       const ::im_proto::GetRecentHisMsgReq* request,
                       ::im_proto::GetRecentHisMsgRsp* response,
                       ::google::protobuf::Closure* done)
    {
        brpc::ClosureGuard rpc_guard(done);
        const std::string& rid = request->request_id();
        response->set_request_id(rid);

        // 1. 获取请求会话 && 消息条数, 差数据库
        const std::string& chatSessionId = request->chat_session_id();
        int64_t msgCount = request->msg_count();
        auto msgList = messageDao_->selectRecentMessages(chatSessionId, msgCount);
        if(msgList.empty()) {
            response->set_success(true);
            return;
        }
        // 2. 构建响应消息列表
        makeRespMsgList(rid, chatSessionId, msgList, response);
    }
    // 获取指定时间范围历史消息
    void GetRangeHisMsg(::google::protobuf::RpcController* controller,
                       const ::im_proto::GetRangeHisMsgReq* request,
                       ::im_proto::GetRangeHisMsgRsp* response,
                       ::google::protobuf::Closure* done)
    {
        brpc::ClosureGuard rpc_guard(done);
        const std::string& rid = request->request_id();
        response->set_request_id(rid);   

        // 1. 获取请求会话 && 时间范围, 查数据库
        const std::string& chatSessionId = request->chat_session_id();
        const boost::posix_time::ptime stime = boost::posix_time::from_time_t(request->start_time());
        const boost::posix_time::ptime etime = boost::posix_time::from_time_t(request->over_time());
        auto msgList = messageDao_->selectRangeMessages(chatSessionId, stime, etime);
        if(msgList.empty()) {
            response->set_success(true);
            return;
        }
        // 2. 构建响应消息列表
        makeRespMsgList(rid, chatSessionId, msgList, response);
    }
    // 历史消息搜索
    void HisMsgSearch(::google::protobuf::RpcController* controller,
                       const ::im_proto::HisMsgSearchReq* request,
                       ::im_proto::HisMsgSearchRsp* response,
                       ::google::protobuf::Closure* done)
    {
        brpc::ClosureGuard rpc_guard(done);
        const std::string& rid = request->request_id();
        response->set_request_id(rid);

        // 1. 获取会话, 搜索关键字 --> es搜索消息
        const std::string& chatSessionId = request->chat_session_id();
        const std::string& key = request->search_key();
        auto msgList = esMessage_->search(key, chatSessionId);
        if(msgList.empty()) {
            response->set_success(true);
            return;
        }
        // 2. rpc获取消息对应的用户信息
        GPUserMap userInfoMap;
        if(_getUser(rid, msgList, userInfoMap) == false) {
            LOG_ERROR("{}: 批量用户数据获取失败！", rid);
            return errResp(response, "批量用户数据获取失败!");
        }
        // 3. 构建搜索结果响应
        response->set_success(true);
        for(const auto& msg : msgList) {
            auto respMsg = response->add_msg_list();
            // makeRespMsgBaseInfo(chatSessionId, userInfoMap.at(msg.userId()), msg, respMsg);
            respMsg->set_message_id(msg.messageId());
            respMsg->set_chat_session_id(chatSessionId);
            respMsg->set_timestamp(boost::posix_time::to_time_t(msg.createTime()));
            respMsg->mutable_sender()->CopyFrom(userInfoMap.at(msg.userId()));
            respMsg->mutable_message()->set_message_type(static_cast<im_proto::MessageType>(msg.messageType()));
            respMsg->mutable_message()->mutable_string_message()->set_content(msg.content());
        }
    }

public:
    void onMessage(const char* body, size_t len)
    {
        // 1. 反序列化消息
        im_proto::MessageInfo messageInfo;
        if(messageInfo.ParseFromArray(body, len) == false) {
            LOG_ERROR("消费消息时反序列化失败!");
            return;
        }
        // 2. 消息类型分别处理
        std::string fileId, filename, content;
        fileId.resize(32);
        uint64_t fileSize = 0;
        switch (messageInfo.message().message_type()) {
            case im_proto::MessageType::STRING:
            {
                // 新增es文本消息索引
                content = messageInfo.message().string_message().content();
                if(esMessage_->appendData(
                    messageInfo.chat_session_id(),
                    messageInfo.message_id(), 
                    messageInfo.sender().user_id(), 
                    messageInfo.timestamp(),
                    content) == false) 
                {
                    LOG_ERROR("新增文本消息索引失败!");
                    return;
                }
            }
            break;
            // 文件消息则将文件上传给文件子服务进行存储
            case im_proto::MessageType::IMAGE:
            {
                const auto& imageMsg = messageInfo.message().image_message();
                if(_putFile("", imageMsg.image_content(), imageMsg.image_content().size(), fileId.data()) == false) {
                    LOG_ERROR("上传图片消息到文件子服务失败!");
                    return;
                }
            }
            break;
            case im_proto::MessageType::FILE:
            {
                const auto& fileMsg = messageInfo.message().file_message();
                filename = fileMsg.file_name();
                fileSize = fileMsg.file_size();
                if(_putFile(filename, fileMsg.file_contents(), fileSize, fileId.data()) == false) {
                    LOG_ERROR("上传文件消息到文件子服务失败!");
                    return;
                }
            }
            break;
            case im_proto::MessageType::SPEECH:
            {
                const auto& speechMsg = messageInfo.message().speech_message();
                if(_putFile("", speechMsg.file_contents(), speechMsg.file_contents().size(), fileId.data()) == false) {
                    LOG_ERROR("上传语音消息到文件子服务失败!");
                    return;
                }
            }
            break;
            default:
                LOG_ERROR("消息类型错误！");
                return;
        }
        
        // fileId不一定有数据, 无数据的话把空间释放了,
        // 否则后续getFile的时候, empty无法捕捉空字符串
        fileId.resize(strlen(fileId.c_str()));  

        // 3. 持久化        
        entity::Message message(messageInfo.message_id(),
            messageInfo.chat_session_id(),
            messageInfo.sender().user_id(),
            messageInfo.message().message_type(),
            boost::posix_time::from_time_t(messageInfo.timestamp())
        );
        message.setFileId(fileId);
        message.setFilename(filename);
        message.setFileSize(fileSize);
        message.setContent(content);
        if(messageDao_->insert(message) == false) {
            LOG_ERROR("向数据库插入新的消息记录失败！");
            return;
        }
        LOG_DEBUG("从消息队列消费并持久化消息-{}", content.empty() ? "文件类消息" : content);
    }
private:
    template<class T>
    void makeRespMsgList(const std::string& rid, const std::string& chatSessionId, 
        const vector<entity::Message>& msgList, T* response)
    {
        // 1. rpc通信获取文件消息数据
        GPFileMap fileDataMap;
        if(_getFile(rid, msgList, fileDataMap) == false) {
            LOG_ERROR("{}: 批量文件数据下载失败!", rid);
            return errResp(response, "批量文件数据下载失败");
        }
        // 2. rpc通信获取消息所属用户信息
        GPUserMap userInfoMap;
        if(_getUser(rid, msgList, userInfoMap) == false) {
            LOG_ERROR("{}: 批量用户信息获取失败!", rid);
            return errResp(response, "批量用户信息获取失败");
        }
        response->set_success(true);
        // 3. 构造响应消息集合
        for(const auto& msg : msgList) {
            auto respMsg = response->add_msg_list();
            makeRespMsgBaseInfo(chatSessionId, userInfoMap.at(msg.userId()), msg, respMsg);
            auto mutableMsg = respMsg->mutable_message();
            switch (msg.messageType()) {
                case im_proto::MessageType::STRING:
                    mutableMsg->mutable_string_message()->set_content(msg.content());
                    break;
                case im_proto::MessageType::IMAGE:
                    mutableMsg->mutable_image_message()->set_file_id(msg.fileId());
                    mutableMsg->mutable_image_message()->set_image_content(fileDataMap.at(msg.fileId()).file_content());
                    break;
                case im_proto::MessageType::FILE:
                    mutableMsg->mutable_file_message()->set_file_id(msg.fileId());
                    mutableMsg->mutable_file_message()->set_file_name(msg.filename());
                    mutableMsg->mutable_file_message()->set_file_contents(fileDataMap.at(msg.fileId()).file_content());
                    mutableMsg->mutable_file_message()->set_file_size(msg.fileSize());
                    break;
                case im_proto::MessageType::SPEECH:
                    mutableMsg->mutable_speech_message()->set_file_id(msg.fileId());
                    mutableMsg->mutable_speech_message()->set_file_contents(fileDataMap.at(msg.fileId()).file_content());
                    break;
                default:
                    LOG_ERROR("错误的消息类型");
                    break;
            }
        }   
    }

    bool _getUser(const std::string& rid, const vector<entity::Message>& msgList, GPUserMap& userInfoMap)
    {
        std::unordered_set<std::string> userIdSet;
        for(const auto& msg : msgList) {
            userIdSet.emplace(msg.userId());
        }

        auto channel = rpcServiceMgr_->choose(userServiceName_);
        if(!channel) {
            LOG_ERROR("用户管理子服务节点({})不存在!", userServiceName_);
            return false;
        }  
        im_proto::GetMultiUserInfoReq req;
        req.set_request_id(rid);
        for(const auto& userId : userIdSet) {
            req.add_users_id(userId);
        }

        brpc::Controller ctl;
        im_proto::GetMultiUserInfoRsp resp;
        im_proto::UserService_Stub stub(channel.get());
        stub.GetMultiUserInfo(&ctl, &req, &resp, nullptr);
        if(ctl.Failed() || !resp.success()) {
            LOG_ERROR("获取批量用户信息失败: {}", ctl.ErrorText());
            return false;
        } 
        userInfoMap = resp.users_info();
        return true;
    }
    bool _getFile(const std::string& rid, const vector<entity::Message>& msgList, GPFileMap& fileDataMap)
    {
        std::unordered_set<std::string> fileIdSet;
        for(const auto& msg : msgList) {
            if(msg.fileId().empty()) { 
                continue;
            }
            fileIdSet.emplace(msg.fileId());
        }
        auto channel = rpcServiceMgr_->choose(fileServiceName_);
        if(!channel) {
            LOG_ERROR("文件子服务节点({})不存在!", fileServiceName_);
            return false;
        }  
        im_proto::GetMultiFileReq req;
        req.set_request_id(rid);
        for(const auto& fileId : fileIdSet) {
            req.add_file_id_list(fileId);
        }
        brpc::Controller ctl;
        im_proto::GetMultiFileRsp resp;
        im_proto::FileService_Stub stub(channel.get());
        stub.GetMultiFile(&ctl, &req, &resp, nullptr);
        if(ctl.Failed() || !resp.success()) {
            LOG_ERROR("下载批量文件失败: {}", ctl.ErrorText());
            return false;
        } 
        fileDataMap = resp.file_data();
        return true;
    }
    void makeRespMsgBaseInfo(const std::string& chatSessionId, const im_proto::UserInfo& userInfo, 
        const entity::Message& msg, im_proto::MessageInfo* respMsg)
    {
        respMsg->set_message_id(msg.messageId());
        respMsg->set_chat_session_id(chatSessionId);
        respMsg->set_timestamp(boost::posix_time::to_time_t(msg.createTime()));
        respMsg->mutable_sender()->CopyFrom(userInfo);
        respMsg->mutable_message()->set_message_type(static_cast<im_proto::MessageType>(msg.messageType()));
    }
    bool _putFile(const std::string& filename, const std::string& content, const uint64_t fileSize, char* fileId)
    {
        // 获取文件服务的rpc通信信道
        auto channel = rpcServiceMgr_->choose(fileServiceName_);
        if(!channel) {
            LOG_ERROR("文件子服务节点({})不存在!", fileServiceName_);
            return false;
        }  
        // 构造上传单文件rpc请求
        im_proto::PutSingleFileReq req;
        req.mutable_file_data()->set_file_name(filename);
        req.mutable_file_data()->set_file_content(content);
        req.mutable_file_data()->set_file_size(fileSize);
    
        // 2-2 通过信道发起服务请求
        brpc::Controller ctl;
        im_proto::PutSingleFileRsp resp;
        im_proto::FileService_Stub stub(channel.get());
        stub.PutSingleFile(&ctl, &req, &resp, nullptr);
        if(ctl.Failed() || !resp.success()) {
            LOG_ERROR("上传文件失败: {}", ctl.ErrorText());
            return false;
        }
        strcpy(fileId, resp.file_info().file_id().c_str());
        return true;
    }
    template<class Resp>
    void errResp(Resp* response, const std::string& errmsg) {
        response->set_success(false);
        response->set_errmsg(errmsg);
    }
private:
    shared::dal::MessageSearchPtr esMessage_;    // 消息搜索的操作对象
    shared::dal::MessageDaoPtr messageDao_;     // 持久化数据的操作对象

    shared::infra::RpcServiceManagerPtr rpcServiceMgr_;  // rpc的服务管理对象

    std::string fileServiceName_;       // 文件子服务名称, 通过rpc获取文件消息
    std::string userServiceName_;       // 用户管理子服务名称, 通过rpc获取用户信息
};

class MsgStorageServer
{
public:
    MsgStorageServer(const shared::dal::ESClientPtr& esClient,
        const shared::dal::OdbPtr& mysqlClient,
        const shared::infra::MQClientPtr mqClient,
        const std::shared_ptr<brpc::Server>& rpcServer,
        const shared::infra::DiscoveryPtr& rpcSvrDisc, 
        const shared::infra::RegistrantPtr& rpcSvrRegist)
        : esClient_(esClient), mysqlClient_(mysqlClient), mqClient_(mqClient)
        , rpcServer_(rpcServer), rpcSvrDisc_(rpcSvrDisc), rpcSvrRegist_(rpcSvrRegist) {}
    ~MsgStorageServer() {}
    void start() { rpcServer_->RunUntilAskedToQuit(); }
private:
    shared::dal::ESClientPtr esClient_;     // 支撑消息存储业务功能的搜索引擎句柄
    shared::dal::OdbPtr mysqlClient_;       // 消息存储的持久化句柄
    shared::infra::MQClientPtr mqClient_;    // 消息队列客户端句柄

    // 这三个对象都不是直接面向业务功能的, 而是为业务功能提供服务支撑
    // 因此服务没有结束, 这三个对象就不能释放, 于是放在MsgStorageServer中
    std::shared_ptr<brpc::Server> rpcServer_;        // rpc服务器
    shared::infra::DiscoveryPtr rpcSvrDisc_;         // rpc服务发现对象
    shared::infra::RegistrantPtr rpcSvrRegist_;      // rpc服务注册对象
};

class MsgStorageServerBuilder
{
public:
    // 构造搜索引擎对象
    void makeESObject(const std::vector<std::string>& hostList)
    {
        esClient_ = shared::dal::ESClientFactory::create(hostList);
    }
    // 构造消息队列客户端对象
    void makeRabbitMQObject(const std::string& user, const std::string& passwd, const std::string& host,
        const std::string& exchange, const std::string& queue, const std::string& bindingKey)
    {
        mqClient_ = std::make_shared<shared::infra::MQClient>(user, passwd, host);
        mqClient_->declareComponents(exchange, queue, bindingKey);
        queue_ = queue;
    }
    // 构造odb-mysql客户端对象
    void makeMysqlObject(const std::string& MsgStorage,  const std::string& passwd, 
        const std::string& host, const std::string& db,
        const std::string& cset, uint16_t port, uint16_t connectPoolCount)
    {
        mysqlClient_ = shared::dal::ODBFactory::create(
            MsgStorage, passwd, host, db, cset, port, connectPoolCount
        );
    }
    // 构造Rpc服务注册客户端对象
    void makeRpcSvrRegistrant(const std::string& host, const std::string& serviceName, const std::string& accessHost)
    {
        rpcSvrRegist_ = std::make_shared<shared::infra::Registrant>(host);
        rpcSvrRegist_->registry(serviceName, accessHost);
    }
    // 构造Rpc服务发现客户端对象
    void makeRpcSvrDiscoverer(const std::string& host, const std::string& baseServiceName, 
                    const std::string& fileServiceName, const std::string& userServiceName)
    {
        fileServiceName_ = fileServiceName;     
        userServiceName_ = userServiceName;     
        rpcServiceMgr_ = std::make_shared<shared::infra::RpcServiceManager>();
        rpcServiceMgr_->declared(fileServiceName);
        rpcServiceMgr_->declared(userServiceName);
        auto putCb = std::bind(
            &shared::infra::RpcServiceManager::onServiceOnline, rpcServiceMgr_.get(),
            std::placeholders::_1, std::placeholders::_2
        );
        auto delCb = std::bind(
            &shared::infra::RpcServiceManager::onServiceOffline, rpcServiceMgr_.get(),
            std::placeholders::_1, std::placeholders::_2
        );
        rpcSvrDisc_ = std::make_shared<shared::infra::Discovery>(host, baseServiceName, putCb, delCb);
    }
    // 构造RPC服务器对象
    void makeRpcServer(uint16_t port, int32_t timeout, uint8_t numThreads)
    {
        if(!esClient_) {
            LOG_ERROR("还未初始化ES搜索引擎模块!");
            abort();
        }
        if (!mysqlClient_) {
            LOG_ERROR("还未初始化mysql持久化模块!");
            abort();
        }
        if(!mqClient_) {
            LOG_ERROR("还未初始化rabbitMQ消息队列模块!");
            abort();
        }
        if (!rpcServiceMgr_) {
            LOG_ERROR("还未初始化rpc服务管理模块!");
            abort();
        }
        rpcServer_ = std::make_shared<brpc::Server>();
        
        // 向rpc服务器添加服务实现类
        MsgStorageeServiceImpl* msgStorageeServiceImpl = new MsgStorageeServiceImpl(
            esClient_, mysqlClient_, rpcServiceMgr_, fileServiceName_, userServiceName_
        );
        if(rpcServer_->AddService(msgStorageeServiceImpl, brpc::ServiceOwnership::SERVER_OWNS_SERVICE) == -1) {
            LOG_ERROR("添加Rpc服务失败!");
            abort();
        }
        // 启动服务器
        brpc::ServerOptions options;    
        options.idle_timeout_sec = timeout;  
        options.num_threads = numThreads;        
        if(rpcServer_->Start(port, &options) == -1) {
            LOG_ERROR("Rpc服务启动失败!");
            abort();
        }

        // 订阅消息队列
        auto onMessageCB = std::bind(&MsgStorageeServiceImpl::onMessage, 
            msgStorageeServiceImpl, std::placeholders::_1, std::placeholders::_2);
        mqClient_->consume(queue_, onMessageCB);
    }
    MsgStorageServerPtr build()
    {
        if(!rpcSvrDisc_) {
            LOG_ERROR("还未初始化服务发现模块!");
            abort();
        }
        if (!rpcSvrRegist_) {
            LOG_ERROR("还未初始化服务注册模块!");
            abort();
        }
        if (!rpcServer_) {
            LOG_ERROR("还未初始化RPC服务器模块!");
            abort();
        }
        return std::make_shared<MsgStorageServer>(
            esClient_, mysqlClient_, mqClient_, rpcServer_, rpcSvrDisc_, rpcSvrRegist_
        );
    }
private:
    shared::dal::ESClientPtr esClient_;     // 支撑消息存储业务功能的搜索引擎句柄
    shared::dal::OdbPtr mysqlClient_;       // 消息存储的持久化句柄

    std::shared_ptr<brpc::Server> rpcServer_;               // rpc服务器
    shared::infra::RpcServiceManagerPtr rpcServiceMgr_;     // rpc的服务管理对象
    shared::infra::RegistrantPtr rpcSvrRegist_;             // rpc服务注册对象
    shared::infra::DiscoveryPtr rpcSvrDisc_;                // rpc服务发现对象
    
    shared::infra::MQClientPtr mqClient_;    // 消息队列客户端句柄
    std::string queue_;                      // 订阅队列消息

    std::string fileServiceName_;           // 文件子服务名称, 通过rpc获取文件消息数据
    std::string userServiceName_;           // 用户管理子服务名称, 通过rpc获取用户信息
};



}